package com.ebupt.migu.music.push.service.impl;

import cn.hutool.core.io.file.FileReader;
import cn.hutool.core.io.file.FileWriter;
import com.alibaba.fastjson.JSONObject;
import com.ebupt.migu.common.pojo.ErrorReport;
import com.ebupt.migu.common.pojo.UserGroupStatus;
import com.ebupt.migu.common.push.PushContentObject;
import com.ebupt.migu.common.push.PushPolicyObject;
import com.ebupt.migu.common.push.PushRequestBody;
import com.ebupt.migu.common.util.DateUtil;
import com.ebupt.migu.common.util.HttpUtils;
import com.ebupt.migu.common.util.RedisLock;
import com.ebupt.migu.music.common.constants.StrategyConstants;
import com.ebupt.migu.music.push.entity.PushResult;
import com.ebupt.migu.music.push.mapper.PushTaskMapper;
import com.ebupt.migu.music.push.service.KafkaProducerService;
import com.ebupt.migu.music.push.service.ProductPush;
import com.ebupt.migu.music.push.service.PushTaskService;
import com.ebupt.migu.restapi.DownloadByFtp;
import com.ebupt.migu.restapi.GroupDetailDel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.Cursor;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ScanOptions;
import org.springframework.data.redis.support.atomic.RedisAtomicLong;
import org.springframework.stereotype.Service;
import org.springframework.web.util.HtmlUtils;

import javax.annotation.Resource;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.IOException;
import java.lang.reflect.Field;
import java.net.URLEncoder;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * Created by Intellij IDEA. User: liuyangyang@cd.ebupt.com Date: 2019/10/31
 */
@Slf4j
@Service
public class PushTaskServiceImpl implements PushTaskService {

    @Value("${push.ip}")
    private String PUSH_IP;
    @Value("${push.app_id}")
    private String PUSH_APP_ID;
    @Value("${push.app_key}")
    private String PUSH_APP_KEY;
    @Value("${push.app_secret}")
    private String PUSH_APP_SECRET;
    @Value("${push.push_url}")
    private String PUSH_URL;
    @Value("${push.is_sync}")
    private boolean IS_SYNC;
    @Value("${push.send_size}")
    private String SEND_SIZE;
    @Value("${path.image_server_path}")
    private String IMAGE_SERVER_PATH;
    @Value("${message.send_url}")
    private String SEND_MESSAGE_URL;
    @Value("${label.post_url}")
    private String postUrl;

    @Resource
    private PushTaskMapper pushTaskMapper;

    @Resource
    private RedisTemplate redisTemplate;

    private static String USER_CODE = "user_info_uid_";

    @Autowired
    private KafkaProducerService producerService;
    @Resource
    private ProductPush productPush;

    /**
     * 用户信息
     */
    private final String _REDIS_KEY = "_user_info";

    /**
     * 立即Push策略
     */
    @Override
    public void pushNow() {
        long start = System.currentTimeMillis();
        Date date = new Date();
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat(DateUtil.DATE_TIME_FORMAT_YYYY_MM_DD_HH_MI_SS);
        SimpleDateFormat simpleDateFormatTaskDay = new SimpleDateFormat(DateUtil.DATE_FORMAT_YYYYMMDD);
        // 当前时间
        String now = simpleDateFormat.format(date);
        // task id拼接的时间
        String taskDay = simpleDateFormatTaskDay.format(date);
        // 查询需要立即发送的短信策略列表
        List<HashMap<String, Object>> strategies = pushTaskMapper.queryPushNow();
        if (null == strategies || strategies.isEmpty()) {
            log.info("当前不存在需要立即PUSH的策略");
            return;
        }
        log.info("当前需要立即执行的PUSH策略共有:{}条", strategies.size());
        Map<String, Object> strategy = strategies.get(0);
        String strategyName = strategy.get("strategyName").toString();
        // 获取策略ID更新相应状态
        String strategyId = strategy.get("strategyId").toString();
        Map<String, Object> map = new HashMap<>();
        // 根据策略关联人群状态判断是否执行下发操作
        if (isSend(strategy, now)) {
            RedisLock redisLock = new RedisLock(redisTemplate, strategyId);
            if (!redisLock.getLock()) {
                log.info("{}已在其他调度中开始下发，本机终止运营", strategyId);
                return;
            }
            log.info("执行策略Id:{},执行策略名称:{}", strategyId, strategyName);
            // 执行流程
            pro(strategy, map, taskDay, (byte)6);
            PushResult pushResult = new PushResult();
            pushResult.setCode(0);
            pushResult.setIsSync(0);
            pushResult.setJobType(1);
            pushResult.setPushDate(taskDay);
            pushResult.setStrategyId(strategyId);
            pushResult.setResultBack(0);
            // 获取groupSize
            pushResult.setGroupSize(getGroupSize(strategyId));
            pushResult.setTaskId(strategyId.concat(taskDay));
            pushTaskMapper.createPushResult(pushResult);
        } else {
            log.info("未执行的策略Id:{},名称:{},人群状态异常", strategyId, strategyName);
        }
        // 更新据库标识位
        map.clear();
        map.put("strategyId", strategyId);
        map.put("executing", 1);
        map.put("executeStatus", StrategyConstants.EXECUTE_ENDED);
        log.info("将策略状态修改为已完成");
        pushTaskMapper.updateStrategyStatus(map);
        long end = System.currentTimeMillis();
        log.info("本次push耗时:{}ms", end - start);
    }

    /**
     * 单次Push策略
     */
    @Override
    public void pushOnce() {
        Date date = new Date();
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat(DateUtil.DATE_TIME_FORMAT_YYYY_MM_DD_HH_MI_SS);
        SimpleDateFormat simpleDateFormatTaskDay = new SimpleDateFormat(DateUtil.DATE_FORMAT_YYYYMMDD);
        // task id拼接的时间
        String taskDay = simpleDateFormatTaskDay.format(date);
        // 当前时间
        String now = simpleDateFormat.format(date);

        // 查询需要立即PUSH的策略列表
        List<HashMap<String, Object>> strategies = pushTaskMapper.queryPushOnce(now);
        if (null == strategies || strategies.isEmpty()) {
            log.info("当前不存在指定时间PUSH的策略");
            return;
        }
        log.info("当前需要立即执行的PUSH策略共有:{}条", strategies.size());
        Map<String, Object> strategy = strategies.get(0);
        String strategyName = strategy.get("strategyName").toString();
        // 获取策略ID更新相应状态
        String strategyId = strategy.get("strategyId").toString();
        Map<String, Object> map = new HashMap<>();
        // 根据策略关联人群状态判断是否执行下发操作
        if (isSend(strategy, now)) {
            RedisLock redisLock = new RedisLock(redisTemplate, strategyId);
            if (!redisLock.getLock()) {
                log.info("{}已在其他调度中开始下发，本机终止运营", strategyId);
                return;
            }
            log.info("执行策略Id:{},执行策略名称:{}", strategyId, strategyName);
            // 执行流程
            pro(strategy, map, taskDay, (byte)6);
            PushResult pushResult = new PushResult();
            pushResult.setCode(0);
            pushResult.setIsSync(0);
            pushResult.setJobType(2);
            pushResult.setPushDate(taskDay);
            pushResult.setStrategyId(strategyId);
            pushResult.setResultBack(0);
            pushResult.setGroupSize(getGroupSize(strategyId));
            pushResult.setTaskId(strategyId.concat(taskDay));
            pushTaskMapper.createPushResult(pushResult);
        } else {
            log.info("未执行的策略Id:{},名称:{},人群状态异常", strategyId, strategyName);
        }
        // 更新据库标识位
        map.clear();
        map.put("strategyId", strategyId);
        map.put("executing", 1);
        map.put("executeStatus", StrategyConstants.EXECUTE_ENDED);
        log.info("将策略状态修改为已完成");
        pushTaskMapper.updateStrategyStatus(map);
    }

    /**
     * 循环Push策略
     */
    @Override
    public void pushLoop() throws ParseException {
        Date date = new Date();
        // 年月日时分秒
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat(DateUtil.DATE_TIME_FORMAT_YYYY_MM_DD_HH_MI_SS);
        // 年月日
        SimpleDateFormat simpleDateFormat_day = new SimpleDateFormat(DateUtil.DATE_FORMAT_YYYY_MM_DD);
        // 时分秒
        SimpleDateFormat simpleDateFormat_time = new SimpleDateFormat(DateUtil.DATE_TIME_FORMAT_HH_MI_SS);
        SimpleDateFormat simpleDateFormatTaskDay = new SimpleDateFormat(DateUtil.DATE_FORMAT_YYYYMMDD);
        // task id拼接的时间将策略状态修改为已完成
        String taskDay = simpleDateFormatTaskDay.format(date);
        // 当前年月日时分秒
        String dayTime = simpleDateFormat.format(date);
        // 当前年月日
        String dayOnly = simpleDateFormat_day.format(date);
        // 当前时分秒
        String timeOnly = simpleDateFormat_time.format(date);

        // 查询需要循环发的策略列表
        List<HashMap<String, Object>> strategies = pushTaskMapper.queryPushLoop(dayOnly);
        if (null == strategies || strategies.isEmpty()) {
            log.info("当前不存在循环PUSH的策略");
            return;
        }
        Map<String, Object> strategy = new HashMap<>();
        for (Map<String, Object> s : strategies) {
            String startTime = s.get("startTime").toString();
            String[] starts = startTime.split(" ");
            String start = starts[1];
            // 最近一次执行时间为空且执行时间小于当前时间
            if (s.get("executeTime") == null
                && (simpleDateFormat_time.parse(start).getTime() <= simpleDateFormat_time.parse(timeOnly).getTime())) {
                strategy = s;
                break;
            }
            // 2020-02-22 修改
            if (s.get("executeTime") != null) {
                String executeTime = s.get("executeTime").toString();
                String[] executes = executeTime.split(" ");
                String execute = executes[0];
                // 最近一次执行时间不为空且最后一次执行日期小于今天且执行时间小于当前时间
                if ((simpleDateFormat_day.parse(execute).getTime() < simpleDateFormat_day.parse(dayOnly).getTime())
                    && (simpleDateFormat_time.parse(start).getTime() <= simpleDateFormat_time.parse(timeOnly)
                        .getTime())) {
                    strategy = s;
                    break;
                }
            }
        }
        if (null == strategy.get("strategyName")) {
            log.info("当前不存在循环PUSH的策略");
            return;
        }
        log.info("执行策略：" + strategy.get("strategyName"));
        // 获取策略ID更新相应状态
        String strategyId = strategy.get("strategyId").toString();
        Map<String, Object> map = new HashMap<>();
        // 策略截止日期
        byte executeStatus = 4;
        String endTime = strategy.get("endTime").toString();
        // 截止日期是今天，将策略状态修改成已完成。 不是今天保留进行中
        String[] ends = endTime.split(" ");
        String end = ends[0];
        if (end.equals(dayOnly)) {
            executeStatus = 6;
        }
        // 根据策略关联人群状态判断是否执行下发操作
        if (isSend(strategy, dayTime)) {
            RedisLock redisLock = new RedisLock(redisTemplate, strategyId);
            if (!redisLock.getLock()) {
                log.info("{}已在其他调度中开始下发，本机终止运营", strategyId);
                return;
            }
            log.info("执行策略Id:{},执行策略名称:{}", strategyId, strategy.get("strategyName"));
            // 执行流程
            pro(strategy, map, taskDay, executeStatus);
            PushResult pushResult = new PushResult();
            pushResult.setCode(0);
            pushResult.setIsSync(0);
            pushResult.setJobType(3);
            pushResult.setPushDate(taskDay);
            pushResult.setStrategyId(strategyId);
            pushResult.setResultBack(0);
            pushResult.setGroupSize(getGroupSize(strategyId));
            pushResult.setTaskId(strategyId.concat(taskDay));
            pushTaskMapper.createPushResult(pushResult);
        } else {
            log.info("未执行的策略Id:{},名称:{},人群状态异常", strategyId, strategy.get("strategyName"));
        }
        map.clear();
        map.put("strategyId", strategyId);
        map.put("executing", 1);
        map.put("executeStatus", executeStatus);
        pushTaskMapper.updateStrategyStatus(map);
        log.info("将策略状态修改为已完成");
    }

    /**
     * 间隔Push策略
     */
    @Override
    public void pushInterval() throws ParseException {
        Date date = new Date();
        // 年月日时分秒
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat(DateUtil.DATE_TIME_FORMAT_YYYY_MM_DD_HH_MI_SS);
        // 年月日
        SimpleDateFormat simpleDateFormat_day = new SimpleDateFormat(DateUtil.DATE_FORMAT_YYYY_MM_DD);
        // 时分秒
        SimpleDateFormat simpleDateFormat_time = new SimpleDateFormat(DateUtil.DATE_TIME_FORMAT_HH_MI_SS);
        SimpleDateFormat simpleDateFormatTaskDay = new SimpleDateFormat(DateUtil.DATE_FORMAT_YYYYMMDD);
        // task id拼接的时间
        String taskDay = simpleDateFormatTaskDay.format(date);
        // 当前年月日时分秒
        String dayTime = simpleDateFormat.format(date);
        // 当前年月日
        String dayOnly = simpleDateFormat_day.format(date);
        // 当前时分秒
        String timeOnly = simpleDateFormat_time.format(date);
        // 查询需要循环发送短信的策略列表
        List<HashMap<String, Object>> strategies = pushTaskMapper.queryPushInterval(dayOnly);
        if (null == strategies || strategies.isEmpty()) {
            log.info("当前不存在间隔PUSH的策略");
            return;
        }
        Map<String, Object> strategy = new HashMap<>();
        for (Map<String, Object> s : strategies) {
            // 间隔时间
            int intervalDays = Integer.parseInt(s.get("intervalDays").toString()) + 1;
            String startTime = s.get("startTime").toString();
            String[] starts = startTime.split(" ");
            String start = starts[1];
            // 最近一次执行时间为空且执行时间小于当前时间
            if (s.get("executeTime") == null
                && (simpleDateFormat_time.parse(start).getTime() <= simpleDateFormat_time.parse(timeOnly).getTime())) {
                strategy = s;
                break;
            }
            // 2020-02-22修改
            if (s.get("executeTime") != null) {
                String executeTime = s.get("executeTime").toString();
                String[] executes = executeTime.split(" ");
                String execute = executes[0];
                // 最近一次执行时间不为空且最后一次执行日期小于今天且执行时间小于当前时间
                if ((simpleDateFormat_day.parse(execute).getTime()
                    + intervalDays * 24 * 60 * 60 * 1000 == simpleDateFormat_day.parse(dayOnly).getTime())
                    && (simpleDateFormat_time.parse(start).getTime() <= simpleDateFormat_time.parse(timeOnly)
                        .getTime())) {
                    strategy = s;
                    break;
                }
            }
        }
        if (null == strategy.get("strategyName")) {
            log.info("当前不存在间隔PUSH策略");
            return;
        }
        log.info("执行策略：" + strategy.get("strategyName"));
        // 获取策略ID更新相应状态
        String strategyId = strategy.get("strategyId").toString();
        Map<String, Object> map = new HashMap<>();
        // 策略截止日期
        String endTime = strategy.get("endTime").toString();
        byte executeStatus = 4;
        // 截止日期是今天，将策略状态修改成已完成。 或下一次执行时间超出截止时间，将策略状态修改为已完成
        int intervalDays = Integer.parseInt(strategy.get("intervalDays").toString()) + 1;
        String[] ends = endTime.split(" ");
        String end = ends[0];
        if ((simpleDateFormat_day.parse(dayOnly).getTime() + intervalDays * 24 * 60 * 60 * 1000 > simpleDateFormat_day
            .parse(end).getTime()) || (end.equals(dayOnly))) {
            executeStatus = 6;
        }
        // 根据策略关联人群状态判断是否执行下发操作
        if (isSend(strategy, dayTime)) {
            RedisLock redisLock = new RedisLock(redisTemplate, strategyId);
            if (!redisLock.getLock()) {
                log.info("{}已在其他调度中开始下发，本机终止运营", strategyId);
                return;
            }
            log.info("执行策略Id:{},执行策略名称:{}", strategyId, strategy.get("strategyName"));
            // 执行流程
            pro(strategy, map, taskDay, executeStatus);
            PushResult pushResult = new PushResult();
            pushResult.setCode(0);
            pushResult.setIsSync(0);
            pushResult.setJobType(4);
            pushResult.setPushDate(taskDay);
            pushResult.setStrategyId(strategyId);
            pushResult.setResultBack(0);
            pushResult.setGroupSize(getGroupSize(strategyId));
            pushResult.setTaskId(strategyId.concat(taskDay));
            pushTaskMapper.createPushResult(pushResult);
        } else {
            log.info("未执行的策略Id:{},名称:{},人群状态异常", strategyId, strategy.get("strategyName"));
        }
        map.clear();
        map.put("strategyId", strategyId);
        map.put("executing", 1);
        map.put("executeStatus", executeStatus);
        pushTaskMapper.updateStrategyStatus(map);
    }

    /**
     * 根据策略关联人群状态判断是否执行下发操作
     *
     * @param strategy
     * @param time
     * @return
     */
    private boolean isSend(Map<String, Object> strategy, String time) {
        boolean isSend;
        String strategyId = strategy.get("strategyId").toString();
        String groupId = strategy.get("groupId").toString();
        String status = pushTaskMapper.qryUserGroupStatus(groupId);
        Map<String, Object> map = new HashMap<>();
        map.put("strategyId", strategyId);
        map.put("executeTime", time);
        if (UserGroupStatus.FINISH.getVal().equals(status) || (null != strategy.get("broadcastPush")
            && 1 == Integer.valueOf(String.valueOf(strategy.get("broadcastPush"))))) {
            map.put("executeStatus", StrategyConstants.EXECUTE_ONGOING);
            map.put("executing", 2);
            isSend = true;
        } else {
            // 人群不是已完成状态，不能发送，记录下发失败日志
            ErrorReport errorReport = new ErrorReport();
            errorReport.setGid(groupId).setSid(strategyId).setCreateTime(time)
                .setRemark(UserGroupStatus.getMsg(status));
            pushTaskMapper.insertErrorReport(errorReport);
            isSend = false;
        }
        // 更新
        pushTaskMapper.updateStrategyStatus(map);
        return isSend;
    }

    /**
     * 构造PushContentObject实例
     *
     * @param strategy
     * @param pushContentObject
     */
    public void pushContent(Map<String, Object> strategy, PushContentObject pushContentObject, String uid,
        boolean ist_use, boolean isv_use) {
        // 需要进行HTML解码
        pushContentObject.setPushTitle(ist_use
            ? replaceUserData(HtmlUtils.htmlUnescape(strategy.get("pushTitle").toString()),
                strategy.get("titleDefaultUsername").toString(), uid)
            : HtmlUtils.htmlUnescape(strategy.get("pushTitle").toString()));
        pushContentObject.setPushContent(isv_use
            ? replaceUserData(HtmlUtils.htmlUnescape(strategy.get("pushContent").toString()),
                strategy.get("smsDefaultUsername").toString(), uid)
            : HtmlUtils.htmlUnescape(strategy.get("pushContent").toString()));
        // 默认值是1
        pushContentObject.setPushType(1);
        // 1.前端输入链接以：mgmusic开头的，pushLinkType=100
        // 2.输入链接以http或以https开头的，pushLinkType=9
        // 20210420调整： 所有链接pushLinkType均设置为100， 以http开头的链接通过转码并拼接上固定字符串进行下发
        if (null != strategy.get("pushUrl")) {
            String pushURL = strategy.get("pushUrl").toString();
            pushContentObject.setPushLinkType(100);
            // TODO 新增mgmedia类型链接，用来下发到融媒app
            if (pushURL.startsWith("http") || pushURL.startsWith("https")) {
                try {
                    String encodeURL = URLEncoder.encode(pushURL, "UTF-8");
                    pushURL = "mgmusic://browser?url=" + encodeURL;
                } catch (Exception e) {
                    log.error("链接转码失败", e.getMessage());
                }
            }
            pushContentObject.setPushLinkAddress(pushURL);
        }

        if (null != strategy.get("picDir") && !"".equals(String.valueOf(strategy.get("picDir")))) {
            String picDir = strategy.get("picDir").toString();
            // picDir = IMAGE_SERVER_PATH + picDir.substring(picDir.lastIndexOf("/") + 1);
            pushContentObject.setImgUrl(picDir);
        }
        pushContentObject.setPushCreateTime(System.currentTimeMillis());
        log.info("push content object:{}", pushContentObject.toString());
    }

    public String replaceUserData(String pushContent, String defaultUsername, String uid) {
        String constructionJson = pushContent;
        // 欧用户则替换 无用户则用原来的
        if (!uid.isEmpty()) {
            try {
                String userName = "";
                HashOperations hashOperations = redisTemplate.opsForHash();
                for (int i = 1; i < 81; i++) {
                    // 如果有用户数则替换
                    if (hashOperations.hasKey(USER_CODE + i, uid)) {
                        userName = String.valueOf(hashOperations.get(USER_CODE + i, uid));
                        log.info("匹配到用户名，名称：{}", userName);
                        break;
                    }
                }
                // 如果没有影虎数据则使用默认
                if (userName.isEmpty()) {
                    log.info("匹配不到用户名，使用备用名称：{}", defaultUsername);
                    userName = defaultUsername.isEmpty() ? "用户" : defaultUsername;
                }
                constructionJson = constructionJson.replaceAll("\\$username\\$", userName);
                log.info("替换后的文本：{},uid为{}", constructionJson, uid);

            } catch (Exception e) {
                log.error("用户替换异常", e);
            }
        }
        return constructionJson;
    }

    private Cursor<String> getRedisCursor(RedisTemplate redisTemplate, String key, Integer num) {
        try {
            return redisTemplate.opsForSet().scan(key, ScanOptions.NONE);
        } catch (Exception e) {
            e.printStackTrace();
            try {
                Thread.sleep(3000L);
            } catch (Exception es) {
                log.info("重调获取内容间歇异常" + key);
            }
            int count = num + 1;
            log.info("次数重调获取内容：" + count + ",key:" + key);
            return getRedisCursor(redisTemplate, key, count);
        }
    }

    private boolean getRedisFlag(RedisTemplate redisTemplate, String key, Integer num) {
        try {
            return redisTemplate.opsForSet().size(key).intValue() != 0;
        } catch (Exception e) {
            e.printStackTrace();
            try {
                Thread.sleep(3000L);
            } catch (Exception es) {
                log.info("重调获取大小间歇异常,key:" + key);
            }
            int count = num + 1;
            log.info("次数重调获取大小：" + count + ",key:" + key);
            return getRedisFlag(redisTemplate, key, count);
        }
    }

    private void exeByRedis(String taskId, Map<String, Object> strategy, List<String> whiteUids, List<String> blacks){
        String strategyId = strategy.get("strategyId").toString();
        String title = strategy.get("pushTitle").toString();
        // 黑名单
        RedisAtomicLong black_incr = new RedisAtomicLong(strategyId + "_" + DateUtil.getDateToday() + "_black",
                Objects.requireNonNull(redisTemplate.getConnectionFactory()));
        // 本地- redis客群规模
        RedisAtomicLong bd_incr = new RedisAtomicLong(strategyId + "_" + DateUtil.getDateToday() + "_bd",
                Objects.requireNonNull(redisTemplate.getConnectionFactory()));
        String redisKey = strategy.get("groupId").toString() + "-uid";
        boolean flag = getRedisFlag(redisTemplate, redisKey, 0);
        log.info("下发策略id:" + taskId + "对应人群key" + redisKey + "判断是否通过" + flag);
        if (flag) {
            Set<String> splitUids = new HashSet<>();
            Cursor<String> cursor = getRedisCursor(redisTemplate, redisKey, 0);
            while (cursor.hasNext()) {
                String uid = cursor.next();
                if (!blacks.isEmpty() && blacks.contains(uid)) {
                    // 黑名单用户
                    black_incr.getAndIncrement();
                    continue;
                }
                splitUids.add(uid);
                if (!whiteUids.isEmpty()) {
                    whiteUids.remove(uid);
                }
                if (splitUids.size() == Integer.parseInt(SEND_SIZE)) {
                    bd_incr.addAndGet(splitUids.size());
                    List<String> uids = new ArrayList<>(splitUids);
                    productPush.productToKafka(taskId, strategy, uids, null, title);
                    // 清空
                    splitUids.clear();
                }
            }
            if (!splitUids.isEmpty()) {
                bd_incr.addAndGet(splitUids.size());
                List<String> uids = new ArrayList<>(splitUids);
                productPush.productToKafka(taskId, strategy, uids, null, title);
                // 发送之后清空
                splitUids.clear();
            }
        }
        if (!whiteUids.isEmpty()) {
            productPush.productToKafka(taskId, strategy, whiteUids, null, title);
        }
        // 黑名单
        black_incr.expire(30, TimeUnit.DAYS);
        // 本地
        bd_incr.expire(30, TimeUnit.DAYS);
    }

    private void exeByFile(String taskId, Map<String, Object> strategy, File file, String recordFile, List<String> whiteUids, List<String> blacks){
        String title = strategy.get("pushTitle").toString();
        String strategyId = strategy.get("strategyId").toString();
        // 黑名单
        RedisAtomicLong black_incr = new RedisAtomicLong(strategyId + "_" + DateUtil.getDateToday() + "_black",
                Objects.requireNonNull(redisTemplate.getConnectionFactory()));
        // 本地- redis客群规模
        RedisAtomicLong bd_incr = new RedisAtomicLong(strategyId + "_" + DateUtil.getDateToday() + "_bd",
                Objects.requireNonNull(redisTemplate.getConnectionFactory()));
        try {
            String uid;
            Set<String> splitUids = new HashSet<>();
            BufferedReader reader = new FileReader(file).getReader();
            BufferedWriter writer = new FileWriter(recordFile).getWriter(true);
            log.info("下发策略id:{} -> 人群文件:{}",taskId,file.getName());
            while ((uid = reader.readLine()) != null) {
                if (!"".equals(uid) && !uid.startsWith("GEN_")) {
                    bd_incr.getAndIncrement();
                    if (!blacks.isEmpty() && blacks.contains(uid)) {
                        black_incr.getAndIncrement();
                        continue;
                    }
                    splitUids.add(uid);
                    // 明细记录
                    writer.write(uid + "|^|" + strategyId);
                    writer.newLine();
                    if (!whiteUids.isEmpty()) {
                        whiteUids.remove(uid);
                    }
                    if (splitUids.size() == Integer.parseInt(SEND_SIZE)) {
                        List<String> uids = new ArrayList<>(splitUids);
                        productPush.productToKafka(taskId, strategy, uids, null, title);
                        // 清空
                        splitUids.clear();
                    }
                }
            }
            reader.close();
            if (!splitUids.isEmpty()) {
                List<String> uids = new ArrayList<>(splitUids);
                productPush.productToKafka(taskId, strategy, uids, null, title);
                // 发送之后清空
                splitUids.clear();
            }
            if (!whiteUids.isEmpty()) {
                productPush.productToKafka(taskId, strategy, whiteUids, null, title);
                for (String whiteUid : whiteUids) {
                    // 明细记录
                    writer.write(whiteUid + "|^|" + strategyId);
                    writer.newLine();
                }
            }
            writer.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        // 黑名单
        black_incr.expire(30, TimeUnit.DAYS);
        // 本地
        bd_incr.expire(30, TimeUnit.DAYS);
    }

    /**
     * 全网推送 - 广播模式
     *
     * @param taskId
     * @param strategy
     * @param topic
     * @return
     */
    private void pushToServer(String taskId, Map<String, Object> strategy, String topic) {
        PushContentObject pushContentObject = new PushContentObject();
        pushContent(strategy, pushContentObject, "", false, false);
        PushRequestBody body = new PushRequestBody();
        // policy固定值
        PushPolicyObject policy = new PushPolicyObject();
        policy.setBroadcast(true);
        policy.setIsOffline(true);
        policy.setPushNetWorkType(0);
        body.setPolicy(policy);
        body.setAppId(PUSH_APP_ID);
        body.setTaskId(taskId);
        body.setContent(pushContentObject);
        body.setTopic(topic);
        body.setPushChannel("1");
        body.setIsSync(true);
        if (String.valueOf(strategy.get("sendChannel")).equals("1")) {
            body.setPushAppType("music");
        }
        if (String.valueOf(strategy.get("sendChannel")).equals("2")) {
            body.setPushAppType("media");
        }
        if (String.valueOf(strategy.get("sendChannel")).equals("3")) {
            body.setPushAppType("all");
        }
        producerService.sendMessage(taskId, body);
    }

    /**
     * JavaBean转换Map
     *
     * @param object
     * @return
     */
    public static Map<String, Object> beanToMap(Object object) {
        Map<String, Object> result = new HashMap<>();
        // 获得类的的属性名数组
        Field[] fields = object.getClass().getDeclaredFields();
        try {
            for (Field field : fields) {
                field.setAccessible(true);
                String name = new String(field.getName());
                result.put(name, field.get(object));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return result;
    }

    /**
     * 请求签名 使用场景: 接入方发送HTTP请求时需要对请求做签名校验 签名方式: Map<String, Object> headers = new HashMap<>(); StringBuilder sb = new
     * StringBuilder("MIGU").append("\n"); sb.append(taskId).append("\n"); sb.append(pushChannel).append("\n");
     * sb.append(pushContent).append("\n"); sb.append(timestamp).append("\n"); sb.append(AppSecret).append("\n"); String
     * Authorization = AppKey + ":" + MD5Util.MD5(sb.toString()); headers.put("Authorization", Authorization);
     * headers.put("timestamp", timestamp); 其中: MIGU:固定值 \n:换行转义符 taskId: 任务id,唯一标示 pushChannel: 推送渠道1-个推 当前只支持个推渠道
     * pushContent:传输到推送系统的content.getPushContent() 详见Content Object timestamp:系统时间戳 AppKey :推送系统分配的appKey
     * AppSecret:推送系统分配的密钥 WARN: 接入方系统时间戳跟推送系统需要保持基本一致，防止重放攻击 推送说明: broadcast为true时表示针对app做推送 此时platform生效，表示对手机系统app推送
     * 如果推送列表 broadcast必为false, uid或者cid列表长度大于1。 当前只支持个推推送，所有pushChannel固定为1。 当uid或者cid长度为1时表示对单个用户进行推送。 md5值为小写md5
     */
    private Map<String, String> header(String taskId, PushRequestBody body) {
        Map<String, String> map = new HashMap<>();

        String timestamp = String.valueOf(System.currentTimeMillis());
        StringBuilder sb = new StringBuilder("MIGU").append("\n");
        sb.append(taskId).append("\n");
        sb.append(body.getPushChannel()).append("\n");
        sb.append(body.getContent().getPushContent()).append("\n");
        sb.append(timestamp).append("\n");
        sb.append(PUSH_APP_SECRET).append("\n");
        String Authorization = PUSH_APP_KEY + ":" + DigestUtils.md5Hex(sb.toString());
        map.put("Content-Type", "application/json; charset=utf-8");
        map.put("Authorization", Authorization);
        map.put("timestamp", timestamp);
        return map;
    }

    /**
     * 剔除黑名单用户
     */
    private List<String> removeBlackList(String[] ids) {
        List<String> members = new ArrayList<>();
        for (String id : ids) {
            List<Map<String, String>> blacks = pushTaskMapper.queryRosterMembers(id);
            if (null != blacks && !blacks.isEmpty()) {
                for (Map<String, String> black : blacks) {
                    log.info("黑名单成员：{}", black);
                    // 剔除uuid
                    if (null != black.get("uuid") && !black.get("uuid").equals("")) {
                        log.info("移除UID：{}", black.get("uuid"));
                        members.add(black.get("uuid"));
                    }
                    // 剔除imei
                    if (null != black.get("imei") && !black.get("imei").equals("")) {
                        log.info("移除IMEI：{}", black.get("imei"));
                        members.add(black.get("imei"));
                    }
                    // 剔除idfa
                    if (null != black.get("idfa") && !black.get("idfa").equals("")) {
                        log.info("移除IDFA：{}", black.get("idfa"));
                        members.add(black.get("idfa"));
                    }
                }
            }
        }
        return members;
    }

    /**
     * 添加白名单用户
     */
    private Map<String, List<String>> addWhiteList(String[] ids) {
        Map<String, List<String>> map = new HashMap<>();
        List<String> uids = new ArrayList<>();
        List<String> dids = new ArrayList<>();
        for (String id : ids) {
            List<Map<String, String>> whites = pushTaskMapper.queryRosterMembers(id);
            if (null != whites && !whites.isEmpty()) {
                for (Map<String, String> white : whites) {
                    log.info("白名单成员：{}", white);
                    // 添加uuid
                    if (null != white.get("uuid") && !white.get("uuid").equals("")) {
                        log.info("添加UID：{}", white.get("uuid"));
                        uids.add(white.get("uuid"));
                        continue;
                    }
                    // 添加imei
                    if (null != white.get("imei") && !white.get("imei").equals("")) {
                        log.info("添加IMEI：{}", white.get("imei"));
                        dids.add(white.get("imei"));
                        continue;
                    }
                    // 添加idfa
                    if (null != white.get("idfa") && !white.get("idfa").equals("")) {
                        log.info("添加IDFA：{}", white.get("idfa"));
                        dids.add(white.get("idfa"));
                        continue;
                    }
                }
            }
        }
        if (!uids.isEmpty()) {
            map.put("uids", uids);
        }
        if (!dids.isEmpty()) {
            map.put("dids", dids);
        }
        return map;
    }

    /**
     * 核心流程
     *
     * @param strategy
     * @param map
     * @param taskDay
     */
    private void pro(Map<String, Object> strategy, Map<String, Object> map, String taskDay, byte status) {
        // 判定是否全网推送
        boolean broadcast = null != strategy.get("broadcastPush")
            && 1 == Integer.parseInt(String.valueOf(strategy.get("broadcastPush")));
        String groupId = String.valueOf(strategy.get("groupId"));
        String strategyId = String.valueOf(strategy.get("strategyId"));
        log.info("人群ID: {}", groupId);
        String taskId = strategyId + taskDay;
        log.info("task id:{}", taskId);
        // 2023-10-16 全网推送
        if (broadcast) {
            // 生产消息到kafka
            pushToServer(taskId, strategy, strategy.get("pushTitle").toString());
        } else {
            // 白名单分组： 1-添加白名单
            String[] white = null;
            // 黑名单分组： 1-剔除黑名单
            String[] black = null;
            // 白名单成员
            List<String> whiteUids = new ArrayList<>();
            // 黑名单成员
            List<String> blacks = new ArrayList<>();
            if (null != strategy.get("white")) {
                // 多选白名单
                white = String.valueOf(strategy.get("white")).split(",");
                log.info("白名单组：{}", white.length);
            }
            if (null != strategy.get("black")) {
                // 多选黑名单
                black = String.valueOf(strategy.get("black")).split(",");
                log.info("黑名单组：{}", black.length);
            }
            // 添加白名单用户
            if (white != null && white.length > 0) {
                log.info("添加白名单用户");
                Map<String, List<String>> members = addWhiteList(white);
                if (members.get("uids") != null && !members.get("uids").isEmpty()) {
                    whiteUids = members.get("uids");
                }
            }
            // 剔除黑名单用户
            if (black != null && black.length > 0) {
                log.info("剔除黑名单用户");
                blacks = removeBlackList(black);
            }
            // 创建sftp任务
            String recordFile = create_sftp_data_sync_record(groupId, strategyId);
            // 判断用户群是否属于标签服务平台
            HashMap<String, Object> labelUserGroup = pushTaskMapper.isLabelUserGroup(groupId);
            if (null != labelUserGroup && labelUserGroup.size() > 0) {
                // 执行push推送
                if (redisTemplate.hasKey(groupId + "-uid")) {
                    int MNum = redisTemplate.opsForSet().size(groupId + "-uid").intValue();
                    log.info("uid's num:{}", MNum);
                    exeByRedis(taskId, strategy, whiteUids, blacks);
                } else {
                    // 对接标签服务平台取人群
                    File file = DownloadByFtp.getLabelUserGroup(strategy, postUrl, 3);
                    if (null == file || file.length() == 0) {
                        log.info("当前策略无对应人群");
                        map.clear();
                        map.put("strategyId", strategyId);
                        map.put("executing", 1);
                        map.put("executeStatus", status);
                        pushTaskMapper.updateStrategyStatus(map);
                        return;
                    }
                    exeByFile(taskId, strategy, file, recordFile, whiteUids, blacks);
                    // 对接标签平台客群销毁通知
                    if (file.delete()) {
                        GroupDetailDel.del(strategy);
                    }
                }
            } else {
                // 默认为redis取人群数据
                int MNum = redisTemplate.opsForSet().size(groupId + "-uid").intValue();
                log.info("uid's num:{}", MNum);
                int INum = redisTemplate.opsForSet().size(groupId + "-push-imei").intValue();
                log.info("imei's num:{}", INum);
                if (MNum == 0 && INum == 0) {
                    log.info("当前策略无对应人群");
                    map.clear();
                    map.put("strategyId", strategyId);
                    map.put("executing", 1);
                    map.put("executeStatus", status);
                    pushTaskMapper.updateStrategyStatus(map);
                    log.info("给人群创建人及人群维护人发消息");
                    return;
                }
                exeByRedis(taskId, strategy, whiteUids, blacks);
            }
        }

    }

    /**
     * 获取对应策略id的人群规模
     *
     * @param strategyId
     * @return
     */
    private Integer getGroupSize(String strategyId) {
        String groupSize = pushTaskMapper.getGroupSize(strategyId);
        return StringUtils.isNotBlank(groupSize) ? Integer.parseInt(groupSize) : 0;
    }

    /**
     * 新增sftp文件同步记录
     *
     * @param groupId
     * @param strategyId
     */
    private String create_sftp_data_sync_record(String groupId, String strategyId) {
        String today = DateUtil.getDateToday();
        Map<String, Object> map = new HashMap<>();
        map.put("groupId", groupId);
        map.put("strategyId", strategyId);
        map.put("localPath", "/nfs_shared/fine-operation/push-send-record-label/");
        map.put("fileName", "a_10008_80030_" + today + "_" + strategyId + ".dat");
        pushTaskMapper.create_sftp_data_sync_record(map);
        return map.get("localPath").toString() + map.get("fileName").toString();
    }

    @Override
    public String media(String activityId) {
        SimpleDateFormat simpleDateFormatTaskDay = new SimpleDateFormat(DateUtil.DATE_FORMAT_YYYYMMDD);
        String taskDay = simpleDateFormatTaskDay.format(new Date());
        String taskId = activityId + taskDay;
        // todo 下发
        HashMap<String, String> params = pushTaskMapper.queryStrategyById(activityId);
        JSONObject request = new JSONObject();
        List<String> uid = new ArrayList<>();
        uid.add("35654f7f-d28e-4811-9d9c-5c41be270267");
        uid.add("17562600");
        uid.add("2f0d63d3-4d97-41e4-bb01-b3e0ddf57fe2");
        uid.add("91947172506");
        uid.add("2e76988b-3186-416a-946f-8330a8050495");
        request.put("uid", uid);
        request.put("pushChannel", "1");
        request.put("appId", PUSH_APP_ID);
        request.put("topic", params.get("pushTitle"));
        request.put("pushAppType", "media");
        request.put("isSync", true);
        request.put("taskId", taskId);
        JSONObject content = new JSONObject();
        content.put("imgUrl", params.get("picDir"));
        content.put("pushContent", params.get("pushContent"));
        content.put("pushCreateTime", System.currentTimeMillis());
        content.put("pushLinkAddress", params.get("pushUrl"));
        content.put("pushLinkType", 100);
        content.put("pushTitle", params.get("pushTitle"));
        content.put("pushType", 1);
        request.put("content", content);
        JSONObject policy = new JSONObject();
        policy.put("broadcast", false);
        policy.put("isOffline", true);
        policy.put("pushNetWorkType", 0);
        request.put("policy", policy);

        PushRequestBody body = JSONObject.parseObject(String.valueOf(request), PushRequestBody.class);
        // 转Map
        Map<String, Object> param = beanToMap(body);
        // 生成Header
        Map<String, String> header = header(taskId, body);
        // URL
        String url = PUSH_IP + PUSH_URL;
        log.info("服务端URL: {}", url);
        String response = HttpUtils.doPost(url, param, header);
        log.info("HTTP请求结果：{}", response);
        return response;
    }

    @Override
    public void cid() {
        List<String> taskIds = pushTaskMapper.cids();
        if (null != taskIds && taskIds.size() != 0) {
            taskIds.stream().forEach(id -> {
                if (redisTemplate.hasKey(id)) {
                    HashMap<String, String> map = new HashMap<>();
                    map.put("taskId", id);
                    map.put("cid", String.valueOf(redisTemplate.opsForValue().get(id)));
                    pushTaskMapper.updatePushResult(map);
                }
            });
        }
    }
}
